/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 * 
 *   http://www.apache.org/licenses/LICENSE-2.0
 * 
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
 
package org.apache.linkis.manager.engineplugin.manager.loaders;

import org.apache.linkis.common.exception.ErrorException;
import org.apache.linkis.manager.engineplugin.common.EngineConnPlugin;
import org.apache.linkis.manager.engineplugin.common.loader.entity.EngineConnPluginInfo;
import org.apache.linkis.manager.engineplugin.common.loader.entity.EngineConnPluginInstance;
import org.apache.linkis.manager.engineplugin.common.exception.EngineConnPluginLoadException;
import org.apache.linkis.manager.engineplugin.common.exception.EngineConnPluginNotFoundException;
import org.apache.linkis.manager.engineplugin.manager.classloader.EngineConnPluginClassLoader;
import org.apache.linkis.manager.engineplugin.manager.config.EngineConnPluginLoaderConf;
import org.apache.linkis.manager.engineplugin.manager.loaders.resource.LocalEngineConnPluginResourceLoader;
import org.apache.linkis.manager.engineplugin.manager.loaders.resource.PluginResource;
import org.apache.linkis.manager.engineplugin.manager.utils.EngineConnPluginUtils;
import org.apache.linkis.manager.engineplugin.manager.utils.ExceptionHelper;
import org.apache.linkis.manager.label.entity.engine.EngineTypeLabel;
import org.apache.commons.io.IOUtils;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.BufferedReader;
import java.io.File;
import java.io.FileReader;
import java.io.IOException;
import java.lang.reflect.Constructor;
import java.net.URI;
import java.net.URISyntaxException;
import java.util.*;


public class DefaultEngineConnPluginLoader extends CacheablesEngineConnPluginLoader {

    private static final Logger LOG = LoggerFactory.getLogger(DefaultEngineConnPluginLoader.class);

    private final List<EngineConnPluginsResourceLoader> resourceLoaders = new ArrayList<>();

    private String rootStorePath;

    private String pluginPropsName;

    private static final String PLUGIN_DIR = "plugin";

    public DefaultEngineConnPluginLoader() throws ErrorException {
        //Check store path (is necessary)
        String storePath = EngineConnPluginLoaderConf.ENGINE_PLUGIN_STORE_PATH().getValue();
        if (StringUtils.isBlank(storePath)) {
            ExceptionHelper.dealErrorException(70061, "You should defined ["
                    + EngineConnPluginLoaderConf.ENGINE_PLUGIN_STORE_PATH().key() + "] in properties file", null);
        }
        //The path can be uri
        try {
            URI storeUri = new URI(storePath);
            if(null != storeUri.getScheme()){
                File storeDir = new File(storeUri);
                storePath = storeDir.getAbsolutePath();
            }
        } catch (URISyntaxException e) {
            //Ignore
        } catch (IllegalArgumentException e){
            ExceptionHelper.dealErrorException(70061, "The value:[" + storePath +"] of ["
                    + EngineConnPluginLoaderConf.ENGINE_PLUGIN_STORE_PATH().key() + "] is incorrect", e);
        }
        this.rootStorePath = storePath;
        this.pluginPropsName = EngineConnPluginLoaderConf.ENGINE_PLUGIN_PROPERTIES_NAME().getValue();
        //Prepare inner loaders
//        resourceLoaders.add(new BmlEngineConnPluginResourceLoader());
        resourceLoaders.add(new LocalEngineConnPluginResourceLoader());
    }

    @Override
    protected EngineConnPluginInstance loadEngineConnPluginInternal(EngineConnPluginInfo enginePluginInfo) throws Exception {
        //Build save path
        String savePath = rootStorePath;
        EngineTypeLabel typeLabel = enginePluginInfo.typeLabel();
        if (!savePath.endsWith(String.valueOf(IOUtils.DIR_SEPARATOR))) {
            savePath += IOUtils.DIR_SEPARATOR;
        }
        savePath += typeLabel.getEngineType() + IOUtils.DIR_SEPARATOR + PLUGIN_DIR + IOUtils.DIR_SEPARATOR;
        if (StringUtils.isNoneBlank(typeLabel.getVersion())) {
            savePath += typeLabel.getVersion() + IOUtils.DIR_SEPARATOR;
        }
        //Try to fetch resourceId from configuration
      /*  if (StringUtils.isBlank(enginePluginInfo.resourceId())) {
            String identify = EngineConnPluginLoaderConf.ENGINE_PLUGIN_RESOURCE_ID_NAME_PREFIX() + typeLabel.getEngineType() + "-" + typeLabel.getVersion();
            String resourceIdInConf = BDPConfiguration.get(identify);
            if (StringUtils.isNotBlank(resourceIdInConf)) {
                LOG.info("Fetch resourceId:[" + resourceIdInConf + "] from properties:[" + identify + "]");
                enginePluginInfo.resourceId_$eq(resourceIdInConf);
            }
        }*/
        EngineConnPlugin enginePlugin = null;
        //Load the resource of engine plugin
        PluginResource pluginResource = null;
        for (int i = 0; i < resourceLoaders.size(); i++) {
            PluginResource resource = resourceLoaders.get(i).loadEngineConnPluginResource(enginePluginInfo, savePath);
            if (null != resource) {
                if (null == pluginResource) {
                    pluginResource = resource;
                } else {
                    //Merge plugin resource
                    pluginResource.merge(pluginResource);
                }
            }
        }
        if (null != pluginResource &&
                null != pluginResource.getUrls() && pluginResource.getUrls().length > 0) {
            //Load engine plugin properties file
            Map<String, Object> props = readFromProperties(savePath + pluginPropsName);
            //Build engine classloader
            ClassLoader currentClassLoader = Thread.currentThread().getContextClassLoader();
            EngineConnPluginClassLoader enginePluginClassLoader = new EngineConnPluginClassLoader(pluginResource.getUrls(), currentClassLoader);
            //Load the engine plugin object
            enginePlugin = loadEngineConnPlugin(enginePluginClassLoader, props);
            if(null != enginePlugin){
                //Need to init engine plugin ?
                LOG.info("Init engine conn plugin:[name: " + typeLabel.getEngineType() +
                        ", version: " + typeLabel.getVersion() + "], invoke method init() ");
                initEngineConnPlugin(enginePlugin, props);
                //New another plugin info
                EngineConnPluginInfo newPluginInfo = new EngineConnPluginInfo(typeLabel, pluginResource.getUpdateTime(),
                        pluginResource.getId(), pluginResource.getVersion(), enginePluginClassLoader);
                return new EngineConnPluginInstance(newPluginInfo, enginePlugin);
            }
         }
        throw new EngineConnPluginNotFoundException("No plugin found " + enginePluginInfo.typeLabel().getStringValue() + ", please check your configuration", null);
    }

    /**
     * Init plugin
     * @param plugin plugin instance
     * @param props parameters
     */
    private void initEngineConnPlugin(EngineConnPlugin plugin, Map<String, Object> props) throws EngineConnPluginLoadException {
        try {
            //Only one method
            plugin.init(props);
        }catch(Exception e){
            throw new EngineConnPluginLoadException("Failed to init engine conn plugin instance", e);
        }
    }

    private EngineConnPlugin loadEngineConnPlugin(EngineConnPluginClassLoader enginePluginClassLoader, Map<String, Object> props) throws EngineConnPluginLoadException {
        //First try to load from spi
        EngineConnPlugin enginePlugin = EngineConnPluginUtils.loadSubEngineConnPluginInSpi(enginePluginClassLoader);
        if(null == enginePlugin){
            //Second, try to find plugin class
            String pluginClass = EngineConnPluginUtils.getEngineConnPluginClass(enginePluginClassLoader);
            //Not contain plugin class
            if(StringUtils.isNotBlank(pluginClass)){
                Class<? extends EngineConnPlugin> enginePluginClass = loadEngineConnPluginClass(pluginClass, enginePluginClassLoader);
                return loadEngineConnPlugin(enginePluginClass, enginePluginClassLoader, props);
            }
        }
        return null;
    }

    private EngineConnPlugin loadEngineConnPlugin(Class<? extends EngineConnPlugin> pluginClass, EngineConnPluginClassLoader enginePluginClassLoader, Map<String, Object> props) throws EngineConnPluginLoadException {
        ClassLoader storeClassLoader = Thread.currentThread().getContextClassLoader();
        Thread.currentThread().setContextClassLoader(enginePluginClassLoader);
        try{
            final Constructor<?>[] constructors = pluginClass.getConstructors();
            if(constructors.length == 0){
                throw new EngineConnPluginLoadException("No public constructor in pluginClass [" + pluginClass.getName() + "]", null);
            }
            //Choose the first one
            Constructor<?> constructor = constructors[0];
            Class<?>[] parameters = constructor.getParameterTypes();
            try {
                if (constructor.getParameterCount() == 0) {
                    return (EngineConnPlugin)constructor.newInstance();
                }else if (constructor.getParameterCount() == 1 && parameters[0] == Map.class){
                    return (EngineConnPlugin)constructor.newInstance(props);
                }else{
                    throw new EngineConnPluginLoadException("Illegal arguments in constructor of pluginClass [" + pluginClass.getName() + "]", null);
                }
            }catch(Exception e){
                if(e instanceof EngineConnPluginLoadException){
                    throw (EngineConnPluginLoadException) e;
                }
                throw new EngineConnPluginLoadException("Unable to construct pluginClass [" + pluginClass.getName() + "]", null);
            }
        }finally{
            Thread.currentThread().setContextClassLoader(storeClassLoader);
        }
    }

    private Class<? extends EngineConnPlugin> loadEngineConnPluginClass(String pluginClass, EngineConnPluginClassLoader classLoader) throws EngineConnPluginLoadException {
        try {
            return classLoader.loadClass(pluginClass).asSubclass(EngineConnPlugin.class);
        } catch (ClassNotFoundException e) {
            throw new EngineConnPluginLoadException("Unable to load class:[" + pluginClass + "]", e);
        }
    }

    @SuppressWarnings(value = {"unchecked", "rawtypes"})
    private Map<String, Object> readFromProperties(String propertiesFile) {
        Map<String, Object> map = new HashMap<>();
        Properties properties = new Properties();
        try {
            BufferedReader reader = new BufferedReader(new FileReader(propertiesFile));
            properties.load(reader);
            map = new HashMap<String, Object>((Map) properties);
        }catch(IOException e){
            //Just warn
        }
        return map;
    }
}
